package cn.xuexiyuan.flinkstudy.example;

import cn.xuexiyuan.flinkstudy.entity.Order;
import cn.xuexiyuan.flinkstudy.entity.OrderAggregateResult;
import cn.xuexiyuan.flinkstudy.test.DataFactory;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;

import java.time.Duration;

/**
 * @Description: 交易大屏
 * @Author 左龙龙
 * @Date 21-3-31
 * @Version 1.0
 **/
public class TransactionBigScreen {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<Order> orderDS = env.addSource(new SourceFunction<Order>() {
            boolean runing = true;

            @Override
            public void run(SourceContext<Order> context) throws Exception {
                while (runing) {
                    Order order = DataFactory.createRandomOrder(10, 5);
                    context.collect(order);
                    Thread.sleep(200);
                }
            }

            @Override
            public void cancel() {
                runing = false;
            }
        });

        // 2.transformation
        SingleOutputStreamOperator<Order> watermarks = orderDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(30)).withTimestampAssigner((order, timestamp) -> order.getOrder_time())
        );

        SingleOutputStreamOperator<OrderAggregateResult> aggregate = watermarks.keyBy(Order::getChannel_keyword)
                .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
                .trigger(ContinuousEventTimeTrigger.of(Time.seconds(1)))
                .aggregate(new PriceAggregate(), new WindowResult());


        // 3.sink
        aggregate.print();

        // 4.execute
        env.execute();
    }
}
